package com.atguigu.flink.exec2;

import com.atguigu.flink.pojo.UserBehavior;
import com.atguigu.flink.utils.MyUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * Created by Smexy on 2023/2/2
 *
 *   每隔5分钟输出最近1小时内点击量最多的前N(3)个商品
 *
 *
 *
 *      1.开窗，开一个基于事件时间的 滑动窗口(slide = 5min, size = 1h)
 *      2.第一次聚合： 按照商品id分组，统计窗口内，每个商品的点击量
 *              [8:00,9:00)-1号商品- 点击30
 *              [8:00,9:00)-2号商品- 点击10
 *              [8:00,9:00)-3号商品- 点击20
 *
 *                [8:05,9:05)-1号商品- 点击31
 *               [8:05,9:05)-2号商品- 点击11
 *               [8:05,9:05)-3号商品- 点击21
 *
 *      3.第二次聚合：按照窗口范围分组，排序求前3
 *      4.第二次聚合，在什么时候触发?
 *              在下游获取到 第一次聚合发送的数据后，要获取数据所属窗口的endtime，根据这个时间制定一个定时器。
 *
 *              当下游的水印推进到窗口的结束时间，说明上游早就过了这个时间(水印从上游发送到下游，上游的时间永远比下游早到)，
 *              证明这个时间之前的所有数据已经到达了下游，此时自动进行top3的排序。
 *
 */
public class Demo3_HotItem
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //创建水印生成策略
        WatermarkStrategy<UserBehavior> watermarkStrategy = WatermarkStrategy
            .<UserBehavior>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getTimestamp());

        env.setParallelism(1);

        //读数据
        env.readTextFile("data/UserBehavior.csv").setParallelism(1)
           .map(new MapFunction<String, UserBehavior>()
           {
               @Override
               public UserBehavior map(String line) throws Exception {
                   String[] words = line.split(",");
                   return new UserBehavior(
                       Long.valueOf(words[0]),
                       Long.valueOf(words[1]),
                       Integer.valueOf(words[2]),
                       words[3],
                       //时间戳必须是ms
                       Long.valueOf(words[4]) * 1000
                   );
               }
           }).setParallelism(1)
           .assignTimestampsAndWatermarks(watermarkStrategy)
           //过滤出PV的数据
           .filter(ub -> "pv".equals(ub.getBehavior()))
           .keyBy(UserBehavior::getItemId)
           .window(SlidingEventTimeWindows.of(Time.hours(1),Time.minutes(5)))
           .aggregate(new AggregateFunction<UserBehavior, Integer, HotItem>()
           {
               @Override
               public Integer createAccumulator() {
                   return 0;
               }

               @Override
               public Integer add(UserBehavior value, Integer accumulator) {
                   return accumulator + 1;
               }

               @Override
               public HotItem getResult(Integer accumulator) {
                   return new HotItem(null, null, null, accumulator);
               }

               @Override
               public Integer merge(Integer a, Integer b) {
                   return null;
               }
           }, new ProcessWindowFunction<HotItem, HotItem, Long, TimeWindow>()
           {
               //只执行一次，在窗口触发时执行
               @Override
               public void process(Long key, Context context, Iterable<HotItem> elements, Collector<HotItem> out) throws Exception {
                   //封装计算数据的其他属性
                   HotItem hotItem = elements.iterator().next();
                   TimeWindow window = context.window();
                   hotItem.setStart(window.getStart());
                   hotItem.setEnd(window.getEnd());
                   hotItem.setItemId(key);
                   //向下游输出
                   out.collect(hotItem);
               }
           })
           .keyBy(HotItem::getEnd)
           .process(new KeyedProcessFunction<Long, HotItem, String>()
           {
               //一个集合，用来缓存当前窗口到达的所有数据
               private ListState<HotItem> listState;
               //一个标记，标记是否是这个窗口的第一条数据，如果是，注册定时器
               private ValueState<Boolean> isFirst;

               @Override
               public void open(Configuration parameters) throws Exception {
                   isFirst = getRuntimeContext().getState(new ValueStateDescriptor<>("isFirst", Boolean.class));
                   listState = getRuntimeContext().getListState(new ListStateDescriptor<>("data", HotItem.class));
               }

               @Override
               public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

                   List<HotItem> top3 = StreamSupport.stream(listState.get().spliterator(), true)
                                                        .sorted((h1, h2) -> -h1.getClick().compareTo(h2.getClick()))
                                                        .limit(3)
                                                        .collect(Collectors.toList());

                   String result = top3.stream()
                                        .map(h -> h.getItemId() + ":" + h.getClick())
                                        .collect(Collectors.joining(","));

                   HotItem hotItem = top3.get(0);

                   out.collect(MyUtil.parseTime(new TimeWindow(hotItem.getStart(),hotItem.getEnd())) +"--->top3：" + result);

               }

               //来一个数据，就调用一次
               @Override
               public void processElement(HotItem value, Context ctx, Collector<String> out) throws Exception {

                   TimerService timerService = ctx.timerService();

                   //是第一条，就注册定时器
                   if (isFirst.value() == null){

                       timerService.registerEventTimeTimer(value.getEnd() + 5000);
                       isFirst.update(false);
                   }

                   //把数据放入集合缓存
                   listState.add(value);

               }
           })
           .print().setParallelism(1);

        env.execute();

    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class HotItem {
        //窗口的起始范围
        private Long start;
        private Long end;
        private Long itemId;
        private Integer click;
    }
}
